package flink.utils;

import org.apache.flink.api.java.utils.ParameterTool;

import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;

public class AppConfig {

    private static final String DEFAULT_CONF_FILENAME = "config.properties";
    private static final String PROD_CONF_FILENAME = "prod.properties";
    private static final String TEST_CONF_FILENAME = "test.properties";
    private static final String CLUSTER_MODE = "cluster.mode";
    private static final String SOURCE_KAFKA_BROKER_LIST = "source.kafka.broker";
    private static final String SOURCE_KAFKA_TOPIC = "source.kafka.topic";
    private static final String SOURCE_KAFKA_GROUP_ID = "source.kafka.group.id";
    private static final String SOURCE_KAFKA_PARTITION = "source.kafka.partition";
    private static final String SINK_KAFKA_BROKER_LIST = "sink.kafka.broker";
    private static final String SINK_KAFKA_TOPIC = "sink.kafka.topic";
    private static final String SINK_KAFKA_TOPIC_EXTRACT_FILED = "sink.kafka.topic.extract.filed";
    private static final String SINK_KAFKA_TOPIC_REMOVE_KEY = "sink.kafka.topic.remove.key";
    private static final String SINK_KAFKA_TOPIC_PREFIX = "sink.kafka.topic.prefix";
    private static final String SINK_KAFKA_KEY_SERIALIZE = "sink.kafka.key.serialize";
    private static final String SINK_KAFKA_PARTITION = "sink.kafka.partition";
    private static final String SOURCE_KAFKA_OFFSET = "source.kafka.offset";
    private static final String HIVE_CATALOG_NAME = "hive.catalog.name";
    private static final String HIVE_DATABASE_NAME = "hive.database.name";
    private static final String HIVE_CONF_DIR = "hive.conf.dir";
    private static ParameterTool parameterTool;
    private static Map<String, String> paramMap;

    public static Logger LOG = LoggerFactory.getLogger(AppConfig.class);

    public static ParameterTool init(String propFilePath) {
        try {
            parameterTool =
                    ParameterTool.fromPropertiesFile(
                            AppConfig.class.getClassLoader().getResourceAsStream(propFilePath));
            Preconditions.checkNotNull(parameterTool, "解析配置文件失败,paramMap为空");
            paramMap = parameterTool.toMap();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return parameterTool;
    }

    public static String getSourceKafkaOffset() {
        return paramMap.get(SOURCE_KAFKA_OFFSET);
    }

    public static String getSourceKafkaBrokerList() {
        return paramMap.get(SOURCE_KAFKA_BROKER_LIST);
    }

    public static String getSourceKafkaTopic() {
        return paramMap.get(SOURCE_KAFKA_TOPIC);
    }

    public static String getSourceKafkaGroupId() {
        return paramMap.get(SOURCE_KAFKA_GROUP_ID);
    }

    public static String getSourceKafkaPartition() {
        return paramMap.get(SOURCE_KAFKA_PARTITION);
    }

    public static String getSinkKafkaBrokerList() {
        return paramMap.get(SINK_KAFKA_BROKER_LIST);
    }

    public static String getSinkkafkaTopic() {
        return paramMap.get(SINK_KAFKA_TOPIC);
    }

    public static String getSinkKafkaTopicExtractFiled() {
        return paramMap.get(SINK_KAFKA_TOPIC_EXTRACT_FILED);
    }

    public static String getSinkKafkaTopicPrefix() {
        return paramMap.get(SINK_KAFKA_TOPIC_PREFIX);
    }

    public static String getSinkKafkaKeySerialize() {
        return paramMap.get(SINK_KAFKA_KEY_SERIALIZE);
    }

    public static String getSinkKafkaTopicRemoveKey() {
        return paramMap.get(SINK_KAFKA_TOPIC_REMOVE_KEY);
    }

    public static String getSinkKafkaPartition() {
        return paramMap.get(SINK_KAFKA_PARTITION);
    }

    public static String getHiveCatalogName() {
        return paramMap.get(HIVE_CATALOG_NAME);
    }

    public static String getHiveDatabaseName() {
        return paramMap.get(HIVE_DATABASE_NAME);
    }

    public static String getHiveConfDir() {
        return paramMap.get(HIVE_CONF_DIR);
    }
}
